package com.bigfans.searchservice.listener;

import com.bigfans.framework.es.ElasticTemplate;
import com.bigfans.framework.es.IndexDocument;
import com.bigfans.framework.es.request.CreateDocumentCriteria;
import com.bigfans.framework.kafka.KafkaConsumerBean;
import com.bigfans.framework.kafka.KafkaListener;
import com.bigfans.model.event.order.OrderCreatedEvent;
import com.bigfans.searchservice.api.clients.OrderServiceClient;
import com.bigfans.searchservice.model.Order;
import com.bigfans.searchservice.model.OrderItem;
import com.bigfans.searchservice.schema.mapping.CategoryMapping;
import com.bigfans.searchservice.schema.mapping.OrderMapping;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.CompletableFuture;

@Component
@KafkaConsumerBean
public class OrderListener {

    private Logger logger = LoggerFactory.getLogger(OrderListener.class);

    @Autowired
    private OrderServiceClient orderServiceClient;
    @Autowired
    private ElasticTemplate elasticTemplate;

    @KafkaListener
    public void on(OrderCreatedEvent event){
        try{
            CompletableFuture<Order> orderFuture = orderServiceClient.getOrder(event.getOrderId());
            CompletableFuture<List<OrderItem>> itemFuture = orderServiceClient.getOrderItems(event.getOrderId());
            CompletableFuture.allOf(orderFuture , itemFuture).join();
            Order order = orderFuture.get();
            List<OrderItem> items = itemFuture.get();

            IndexDocument orderDoc = new IndexDocument(order.getId());
            CreateDocumentCriteria createDocumentCriteria = new CreateDocumentCriteria();
            createDocumentCriteria.setIndex(OrderMapping.INDEX);
            createDocumentCriteria.setType(OrderMapping.TYPE);
            createDocumentCriteria.setDocument(orderDoc);
            elasticTemplate.insert(createDocumentCriteria);


        }catch (Exception e) {
            logger.error(e.getMessage());
        }
    }

}
